package org.llc.flink.stream.mysql;

import java.sql.DriverManager;
import java.sql.ResultSet;

import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import com.mysql.jdbc.Connection;
import com.mysql.jdbc.PreparedStatement;

public class Flink2JdbcSource extends
        RichSourceFunction<Tuple4<String, String, Integer, String>> {
    private static final long serialVersionUID = 3334654984018091675L;

    private Connection connect = null;
    private PreparedStatement ps = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        Class.forName("com.mysql.jdbc.Driver");
        connect = (Connection) DriverManager.getConnection("jdbc:mysql://localhost:3306/from?characterEncoding=utf-8&useSSL=false", "root", "316412316");
        ps = (PreparedStatement) connect
                .prepareStatement("select fd_id, doc_subject, doc_status, doc_creator_id from kms_knowledge_base_doc");
    }

    @Override
    public void run(
            SourceContext<Tuple4<String, String, Integer, String>> collect)
            throws Exception {
        ResultSet resultSet = ps.executeQuery();
        while (resultSet.next()) {
            Tuple4<String, String, Integer, String> tuple = new Tuple4<>();
            tuple.setFields(resultSet.getString(1), resultSet.getString(2), resultSet.getInt(3),
                    resultSet.getString(4));
            collect.collect(tuple);
        }

    }

    @Override
    public void cancel() {
        try {
            super.close();
            if (connect != null) {
                connect.close();
            }
            if (ps != null) {
                ps.close();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

}
